package redisson.rest;

import static java.util.stream.Collectors.groupingBy;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.redisson.api.RTimeSeries;
import org.redisson.api.RedissonClient;
import org.redisson.codec.JsonJacksonCodec;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.web.bind.annotation.RestController;

import lombok.extern.slf4j.Slf4j;

@Slf4j
//@RestController
public class TestScheduler implements SmartInitializingSingleton {

	@Autowired
	RedissonClient redissonClient;

	@Override
	public void afterSingletonsInstantiated() {

		ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(5,
				new CustomizableThreadFactory("rpa-worker-"));
		// 入库
		scheduler.scheduleWithFixedDelay(this::flush, 10, 30, TimeUnit.SECONDS);
		// 5分钟成功率计算
		scheduler.scheduleWithFixedDelay(this::rpa5mMonitor, 10, 60, TimeUnit.SECONDS);
		// 15分钟,票据不存在计算
		scheduler.scheduleWithFixedDelay(this::rpa15mMonitor, 10, 60, TimeUnit.SECONDS);
		// 5分钟探活计算
		scheduler.scheduleWithFixedDelay(this::rpa5mProbeMonitor, 10, 60, TimeUnit.SECONDS);

	}

	private void flush() {
		log.info("### 调度 flush");
	}

	private void rpa5mMonitor() {
		log.info("### 调度 rpa5mMonitor");

		RTimeSeries<Map<String, String>, String> ts = redissonClient.getTimeSeries("rpa:probe",
				JsonJacksonCodec.INSTANCE);
		long milli = Instant.now().minus(15, ChronoUnit.MINUTES).toEpochMilli();
		Collection<Map<String, String>> range = ts.range(milli, System.currentTimeMillis());

		Map<String, List<Map<String, String>>> maps = range.stream().collect(groupingBy(map -> map.get("province")));

		maps.forEach((k, v) -> {
			if (v.size() < 10) {
				return;
			}

			// @formatter:off
			long count = v.stream().filter(map -> map.get("success").equals("true")).count();
			double successRatio = new BigDecimal(count)
					.divide(new BigDecimal(v.size()), 2, RoundingMode.HALF_UP)
					.doubleValue();
			// @formatter:on
			if (successRatio > 0.9) {
				log.info("### 调度 rpa5mMonitor 发送消息通知,mock 成功率大于90");
			}
		});
	}

	private void rpa15mMonitor() {
		log.info("### 调度 rpa15mMonitor");
		try {
			TimeUnit.SECONDS.sleep(15);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		log.info("### 调度 rpa15mMonitor 发送消息通知");
	}

	private void rpa5mProbeMonitor() {
		log.info("### 调度 rpa5mProbeMonitor");
	}

}
